package net.wicp.tams.common.es;

import java.io.Serializable;
import java.sql.Connection;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.commons.beanutils.PropertyUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.Validate;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.admin.indices.alias.Alias;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.script.Script;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.googlecode.protobuf.format.JsonFormat;

import lombok.extern.slf4j.Slf4j;
import net.wicp.tams.common.Conf;
import net.wicp.tams.common.apiext.CollectionUtil;
import net.wicp.tams.common.apiext.ReflectAssist;
import net.wicp.tams.common.apiext.StringUtil;
import net.wicp.tams.common.apiext.jdbc.MySqlAssit;
import net.wicp.tams.common.apiext.json.JSONUtil;
import net.wicp.tams.common.apiext.json.SimpleTreeNode;
import net.wicp.tams.common.apiext.json.easyuibean.EasyUINode;
import net.wicp.tams.common.constant.dbType.BinlogType;
import net.wicp.tams.common.constant.dic.YesOrNo;
import net.wicp.tams.common.es.bean.IndexParamsBean;
import net.wicp.tams.common.es.bean.MappingBean;
import net.wicp.tams.common.es.bean.MappingBean.DataTypes;
import net.wicp.tams.common.es.bean.MappingBean.Dynamic;
import net.wicp.tams.common.es.constant.OptTemp;

@Slf4j
public abstract class EsAssit {

	/***
	 * 产生mappingBean，它就可以通过indexCreate创建索引
	 * 
	 * @param mapping
	 * @param conn
	 * @return
	 */
	public static MappingBean proMappingBean(Connection conn, IndexParamsBean mapping) {
		String[][] cols = MySqlAssit.getCols(conn, mapping.getDb(), mapping.getTb(), YesOrNo.yes);
		if (StringUtil.isNotNull(mapping.getDb1()) && StringUtil.isNotNull(mapping.getTb1())) {
			String[][] cols1 = MySqlAssit.getCols(conn, mapping.getDb1(), mapping.getTb1(), YesOrNo.yes);
			String[] nameAry = CollectionUtil.arrayMerge(String[].class, cols[0], cols1[0]);
			String[] typeAry = CollectionUtil.arrayMerge(String[].class, cols[1], cols1[1]);
			cols = new String[][] { nameAry, typeAry };
		}
		if (ArrayUtils.isEmpty(cols) || "_rowkey_".equals(cols[0][0])) {// 没有主键
			log.error("没有列或传入的表没有主键");
			return null;
		}
		String contentjson = EsAssit.packIndexContent(cols[0], cols[1], mapping.buildRelaNodes());

		MappingBean proMappingBean = null;
		try {
			proMappingBean = MappingBean.proMappingBean(contentjson);
		} catch (Exception e) {
			log.error("创建MappingBean失败", e);
			return null;
		}
		return proMappingBean;
	}

	public static CreateIndexRequest indexCreate(Connection conn, IndexParamsBean mapping, String indexName,
			String type, int shardsNum, int replicas, String alias) {
		MappingBean proMappingBean = proMappingBean(conn, mapping);
		if (proMappingBean == null) {
			return null;
		}
		CreateIndexRequest indexCreate = indexCreate(indexName, type, shardsNum, replicas, alias, proMappingBean);
		return indexCreate;
	}

	/****
	 * 得到bulk字符串，
	 * 
	 * @param esData 要更新的数据
	 * @return
	 */
	public static String BulkPack(EsData esData) {
		valid(esData);
		StringBuffer buff = new StringBuffer();
		for (EsObj esObj : esData.getDatasList()) {
			String head = "";
			if (StringUtil.isNull(esObj.getId()) && esData.getAction() == Action.create) {
				head = String.format(OptTemp.actionBulkTempNoId.getValue(), esData.getAction().name(),
						esData.getIndex(), esData.getType());
			} else {
				head = String.format(OptTemp.actionBulkTemp.getValue(), esData.getAction().name(), esData.getIndex(),
						esData.getType(), esObj.getId());
			}
			buff.append(head);
			if (esData.getAction() != Action.delete) {
				String jsonString = JSONObject.toJSONString(esObj.getSourceMap(), false);
				buff.append(jsonString);
				buff.append("\n");
			}
		}
		buff.delete(buff.length() - 2, buff.length() - 1);
		return buff.toString();
	}

	public static String BulkPack(List<EsData> esDatas) {
		StringBuffer buff = new StringBuffer();
		for (EsData esData : esDatas) {
			buff.append(BulkPack(esData));
			buff.append("\n");
		}
		buff.delete(buff.length() - 2, buff.length() - 1);
		return buff.toString();
	}

	public static void valid(EsData esData) {
		Validate.isTrue(esData != null, "参数为空");
		Validate.isTrue(
				esData.getAction() != Action.create || esData.getAction() != Action.index || esData.getDatasCount() > 0,
				"create/index 需要Datas数据");
		Validate.notNull(esData.getAction(), "action不能为空");
		Validate.notBlank(esData.getIndex(), "index不能为空");
		Validate.notBlank(esData.getType(), "type不能为空");
		List<EsObj> datasList = esData.getDatasList();
		Validate.isTrue(datasList.size() > 0, "需要操作的数据");
		if (esData.getAction() == Action.create) {
			return;
		}
		boolean updateSetAll = false;
		if (esData.getAction() == Action.update && esData.hasUpdateSet() && (esData.getUpdateSet().getDoc()
				|| esData.getUpdateSet().getUpsert() || StringUtil.isNotNull(esData.getUpdateSet().getScript()))) {
			updateSetAll = true;
		}
		for (EsObj esObj : datasList) {
			if (esData.getAction() == Action.index || esData.getAction() == Action.delete
					|| esData.getAction() == Action.update) {
				// {"index":"test-tibin90-v1","type":"_doc","id":"","cause":{"type":"illegal_argument_exception","reason":"if
				// _id is specified it must not be empty"},"status":400}
				// Validate.notNull(esObj.getId(), "delete/update/index参数不允许id为null的数据");
				// Validate.notEmpty(esObj.getId(), "delete/update/index参数不允许id为null的数据");
				// //故障修改 20180925 TODO
				if (StringUtil.isNull(esObj.getId())) {
					// continue;//故障修改 20180925 TODO
					log.warn("无id的数据，将会跳过不处理：{}", esObj);
				}
				// Validate.notBlank(esObj.getId(), "delete/update/index参数不允许id为空的数据");
			}
			if (esData.getAction() == Action.update && !updateSetAll) {
				Validate.isTrue(esObj.hasUpdateSet(), "更新操作需要传入updateSet元素");
				Validate.isTrue(
						esObj.getUpdateSet().getDoc() || esObj.getUpdateSet().getUpsert()
								|| StringUtil.isNotNull(esObj.getUpdateSet().getScript()),
						"update操作,doc/upset/script 三者必须要一个起作用");
			}
		}
	}

	/***
	 * 一般对象转为EsData对象
	 * 
	 * @param data
	 * @param action
	 * @param idColName
	 * @param index
	 * @param type
	 * @return
	 */
	@SuppressWarnings({ "rawtypes", "unchecked" })
	public static <T extends Serializable> EsData esData(T data, Action action, String idColName, String index,
			String type, UpdateSet updateSet) {
		Validate.notNull(data, "原始数据不能为空");
		Validate.notNull(action, "action不能为空");
		Validate.notBlank(index, "index不能为空");
		Validate.notBlank(type, "type不能为空");
		EsData.Builder newBuilder = EsData.newBuilder();
		newBuilder.setAction(action);
		newBuilder.setIndex(index);
		newBuilder.setType(type);
		newBuilder.setIsFast(false);
		EsObj.Builder esObjBuilder = EsObj.newBuilder(); // .setId("abcd").putAllSource(newMap).build();
		if (StringUtil.isNotNull(idColName)) {
			if (data instanceof JSONObject) {
				esObjBuilder.setId(((JSONObject) data).getString(idColName));
			} else if (ReflectAssist.isInterface(data.getClass(), "java.util.Map")) {
				Object obj = ((Map) data).get(idColName);
				esObjBuilder.setId((String.valueOf(obj)));
			} else {
				try {
					Object idObj = PropertyUtils.getProperty(data, idColName);
					esObjBuilder.setId(String.valueOf(idObj));
				} catch (Exception e) {
					log.error("取id值出错了", e);
					throw new RuntimeException("取id值出错了", e);
				}
			}
		}
		if (action == Action.update && updateSet != null) {
			esObjBuilder.setUpdateSet(updateSet);
		}

		Map<String, String> map = null;
		if (data instanceof JSONObject) {
			map = ((JSONObject) data).toJavaObject(Map.class);
		} else if (ReflectAssist.isInterface(data.getClass(), "java.util.Map")) {

		} else {
			map = ReflectAssist.convertMapFromBean(data);
		}
		CollectionUtil.filterNull(map, 1);
		esObjBuilder.putAllSource(map);
		newBuilder.addDatas(esObjBuilder.build());
		EsData esData = newBuilder.build();
		valid(esData);
		return esData;
	}

	public static <T extends Serializable> EsData esDataUpdate(T data, String idColName, String index, String type,
			UpdateSet updateSet) {
		return esData(data, Action.update, idColName, index, type, updateSet);
	}

	public static <T extends Serializable> EsData esDataUpdate(String[] datas, String[] fields, String idColName,
			String index, String type, UpdateSet updateSet) {
		Validate.notEmpty(datas);
		Validate.notEmpty(fields);
		Validate.isTrue(fields.length == datas.length);
		JSONObject inputobj = new JSONObject();
		for (int i = 0; i < fields.length; i++) {
			inputobj.put(fields[i], datas[i]);
		}
		return esData(inputobj, Action.update, idColName, index, type, updateSet);
	}

	public static <T extends Serializable> EsData esDataUpsert(String[] datas, String[] fields, String idColName,
			String index, String type) {
		UpdateSet build = UpdateSet.newBuilder().setUpsert(true).build();
		return esDataUpdate(datas, fields, idColName, index, type, build);
	}

	public static <T extends Serializable> EsData esDataDel(String[] datas, String[] fields, String idColName,
			String index, String type) {
		Validate.notEmpty(datas);
		Validate.notEmpty(fields);
		Validate.isTrue(fields.length == datas.length);
		int idIndexOf = ArrayUtils.indexOf(fields, idColName);
		String idvalue = datas[idIndexOf];
		JSONObject input = new JSONObject();
		input.put(idColName, idvalue);
		return esData(input, Action.delete, idColName, index, type, null);
	}

	// 公共处理
	@SuppressWarnings("rawtypes")
	private static void addCommon(DocWriteRequest docWriteRequest, EsObj esObj) {
		if (esObj.hasRelaValue()) {// 路由处理
			RelaValue relaValue = esObj.getRelaValue();
			docWriteRequest.routing(relaValue.getParent());
		}
	}

	@SuppressWarnings({ "rawtypes", "deprecation", "unlikely-arg-type" })
	public static List<DocWriteRequest> packRequest(EsData esData, Map<String, DataTypes> mapping) {
		valid(esData);
		List<DocWriteRequest> retlist = new ArrayList<>();
		switch (esData.getAction()) {
		case create:
			for (EsObj esObj : esData.getDatasList()) {
				IndexRequest createRequest = new IndexRequest(esData.getIndex(), esData.getType());
				createRequest.create(true);
				if (StringUtils.isNotEmpty(esObj.getId())) {
					createRequest.id(esObj.getId());
				}
				createRequest.source(packJson(esObj, mapping), XContentType.JSON);
				addCommon(createRequest, esObj);
				retlist.add(createRequest);
			}
			break;
		case index:
			for (EsObj esObj : esData.getDatasList()) {
				if (StringUtil.isNull(esObj.getId())) {
					continue;
				}
				IndexRequest indexRequest = packIndex(esData, esObj, mapping);
				addCommon(indexRequest, esObj);
				retlist.add(indexRequest);
			}
			break;
		case update:
			for (EsObj esObj : esData.getDatasList()) {
				if (StringUtil.isNull(esObj.getId())) {
					continue;
				}
				UpdateSet updateSetTrue = esObj.hasUpdateSet() ? esObj.getUpdateSet() : esData.getUpdateSet();
				UpdateRequest updateReques = new UpdateRequest(esData.getIndex(), esData.getType(), esObj.getId());
				if (updateSetTrue.getDoc()) {
					IndexRequest indexRequest = packIndex(esData, esObj, mapping);
					updateReques.doc(indexRequest);
					addCommon(updateReques, esObj);
					retlist.add(updateReques);
				} else if (updateSetTrue.getUpsert()) {
					/*
					 * IndexRequest indexRequest = packIndex(esData, esObj, mapping); //
					 * updateReques.upsert(indexRequest); updateReques.docAsUpsert(true);
					 * updateReques.doc(indexRequest);
					 */
					// 全量同步
					IndexRequest createRequest = new IndexRequest(esData.getIndex(), esData.getType(), esObj.getId());
					// createRequest.create(false);//不写会自动判断更新还是新建
					createRequest.source(packJson(esObj, mapping), XContentType.JSON);
					addCommon(createRequest, esObj);
					retlist.add(createRequest);
				} else if (StringUtil.isNotNull(updateSetTrue.getScript())) {
					Script script = new Script(updateSetTrue.getScript());
					updateReques.script(script);
					if (StringUtil.isNotNull(updateSetTrue.getParams())) {
						JSONObject paramObj = JSONObject.parseObject(updateSetTrue.getParams());
						for (String paramkey : paramObj.keySet()) {
							updateReques.addScriptParam(paramkey, paramObj.get(paramObj));
						}
					}
					addCommon(updateReques, esObj);
					retlist.add(updateReques);
				}

			}
			break;
		case delete:
			for (EsObj esObj : esData.getDatasList()) {
				if (StringUtil.isNull(esObj.getId())) {
					continue;
				}
				DeleteRequest deleteRequest = new DeleteRequest(esData.getIndex(), esData.getType(), esObj.getId());
				addCommon(deleteRequest, esObj);
				retlist.add(deleteRequest);
			}
			break;
		default:
			break;
		}
		return retlist;
	}

	/***
	 * 组装关联关系
	 * 
	 * @return
	 */
	public static JSONObject getRelationsJson(List<SimpleTreeNode> nodes) {
		if (CollectionUtils.isEmpty(nodes) || nodes.size() == 1) {
			return null;
		}
		List<EasyUINode> roots = SimpleTreeNode.createTree(nodes);
		if (CollectionUtils.isEmpty(roots) || roots.size() > 1) {
			return null;
		}
		EasyUINode root = roots.get(0);
		JSONObject relaObj = new JSONObject();
		for (SimpleTreeNode node : nodes) {
			EasyUINode curNode = node.getId().equals(root.getId()) ? root : root.getSubById(node.getId());
			List<EasyUINode> childrens = curNode.getChildrens();
			if (CollectionUtils.isNotEmpty(childrens)) {
				if (childrens.size() == 1) {
					relaObj.put(node.getId(), childrens.get(0).getId());
				} else {
					JSONArray arry = new JSONArray();
					for (EasyUINode children : childrens) {
						arry.add(children.getId());
					}
					relaObj.put(node.getId(), arry);
				}
			}
		}
		JSONObject retobj = new JSONObject();
		retobj.put("relations", relaObj);
		retobj.put("eager_global_ordinals", true);// 设置全局加载fielddata
		retobj.put("type", "join");
		return retobj;
	}

	public static String packIndexContent(String[] colName, String[] colType, List<SimpleTreeNode> nodes) {
		Validate.notEmpty(colName);
		Validate.notEmpty(colType);
		Validate.isTrue(colName.length == colType.length);
		JSONObject retobj = new JSONObject();
		for (int i = 0; i < colName.length; i++) {
			BinlogType mysqlType = BinlogType.getByName(colType[i]);
			DataTypes dataTypes = DataTypes.getDataTypesByMysqlType(mysqlType);
			retobj.put(colName[i], dataTypes.getValue());
		}
		if (CollectionUtils.isNotEmpty(nodes)) {
			retobj.put(Conf.get("common.es.assit.rela.key"), getRelationsJson(nodes));
		}
		return JSON.toJSONString(retobj, SerializerFeature.UseSingleQuotes);
	}

	public static String packIndexContent(String[] colName, String[] colType) {
		return packIndexContent(colName, colType, null);
	}

	public static CreateIndexRequest indexCreate(String indexName, String type, int shardsNum, int replicas,
			String alias, MappingBean mappingBean) {
		Validate.notEmpty(indexName);
		Validate.isTrue(shardsNum > 0);
		Validate.isTrue(replicas >= 0);
		Validate.notNull(mappingBean);
		CreateIndexRequest request = new CreateIndexRequest(indexName);
		request.settings(
				Settings.builder().put("index.number_of_shards", shardsNum).put("index.number_of_replicas", replicas));
		// 不格式化会报错：java.lang.IllegalArgumentException
		String jsonString2 = mappingFormat(type, mappingBean);
		request.mapping(type, jsonString2, XContentType.JSON);
		if (StringUtil.isNotNull(alias)) {
			request.alias(new Alias(alias));
		}
		return request;
	}

	public CreateIndexRequest indexCreate(String indexName, String type, int shardsNum, int replicas,
			MappingBean mappingBean) {
		return indexCreate(indexName, type, shardsNum, replicas, null, mappingBean);
	}

	public CreateIndexRequest indexCreate(String indexName, String type, int shardsNum, MappingBean mappingBean) {
		return indexCreate(indexName, type, shardsNum, 0, null, mappingBean);
	}

	private static String mappingFormat(String type, MappingBean mappingBean) {
		SerializeConfig tamsSerializeConfig = SerializeConfig.getGlobalInstance();
		tamsSerializeConfig.put(DataTypes.class, MappingEnumSerializer.instance);
		tamsSerializeConfig.put(Dynamic.class, MappingEnumSerializer.instance);
		String jsonString = JSON.toJSONString(JSONUtil.packParams(type, mappingBean), tamsSerializeConfig);
		String jsonString2 = JSON.toJSONString(JSON.parse(jsonString), true);
		return jsonString2;
	}

	private static IndexRequest packIndex(EsData esData, EsObj esObj, Map<String, DataTypes> mapping) {
		IndexRequest indexRequest = new IndexRequest(esData.getIndex(), esData.getType(), esObj.getId());
		indexRequest.create(false);
		indexRequest.source(packJson(esObj, mapping), XContentType.JSON);
		return indexRequest;
	}

	private static String packJson(EsObj esObj, Map<String, DataTypes> mapping) {
		JSONObject tempobj = new JSONObject();
		Map<String, String> sourceMap = esObj.getSourceMap();
		for (String key : sourceMap.keySet()) {
			DataTypes dataTypes = mapping.get(key);
			if (dataTypes != null) {
				try {
					Object obj = dataTypes.getConvertObj(sourceMap.get(key));
					if (obj != null) {// 20190822 不支持的类型，一般不会走到这来，除非特殊需求，如：str->date
						tempobj.put(key, obj);
					} else {
						log.error(
								"---------------------------处理失败，需要协查，type:{},key:{},value:{}-----------------------------",
								dataTypes.name(), key, sourceMap.get(key));
					}
				} catch (Throwable e) {// 20190822 不支持的类型，一般不会走到这来，除非特殊需求，如：str->date
					log.error(
							"---------------------------处理失败，需要协查，type:{},key:{},value:{}-----------------------------",
							dataTypes.name(), key, sourceMap.get(key));
				}
			}
		}
		// 关联关系处理
		if (esObj.hasRelaValue()) {
			if (esObj.getRelaValue().getParent().isEmpty()) {// 根路径
				tempobj.put(Conf.get("common.es.assit.rela.key"), esObj.getRelaValue().getName());
			} else {
				String json = new JsonFormat().printToString(esObj.getRelaValue());
				JSONObject relaObj = JSONObject.parseObject(json);
				tempobj.put(Conf.get("common.es.assit.rela.key"), relaObj);
			}
		}
		String jsonString = tempobj.toJSONString();
		return jsonString;
	}

}
